package com.ihr360.job.core.repository;

import com.ihr360.job.core.BatchStatus;
import com.ihr360.job.core.JobParameters;
import com.ihr360.job.core.entity.JobExecution;
import com.ihr360.job.core.entity.JobInstance;
import com.ihr360.job.core.entity.StepExecution;
import com.ihr360.job.core.entity.StepExecutionLog;
import com.ihr360.job.core.item.ExecutionContext;
import com.ihr360.job.core.logger.CalcLog;
import com.ihr360.job.core.repository.dao.ExecutionContextDao;
import com.ihr360.job.core.repository.dao.JobExecutionDao;
import com.ihr360.job.core.repository.dao.JobInstanceDao;
import com.ihr360.job.core.repository.dao.StepExecutionDao;
import com.ihr360.job.core.repository.dao.StepExecutionLogDao;
import com.ihr360.job.core.scope.context.SimpleStepContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import javax.batch.operations.JobRestartException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * <p>
 * Implementation of {@link JobRepository} that stores JobInstances,
 * JobExecutions, and StepExecutions using the injected DAOs.
 * <p>
 *
 * @author Lucas Ward
 * @author Dave Syer
 * @author Robert Kasanicky
 * @author David Turanski
 * @see JobRepository
 * @see JobInstanceDao
 * @see JobExecutionDao
 * @see StepExecutionDao
 */
public class SimpleJobRepository implements JobRepository {
    private static final Logger logger = LoggerFactory.getLogger(SimpleJobRepository.class);
    private JobInstanceDao jobInstanceDao;

    private JobExecutionDao jobExecutionDao;

    private StepExecutionDao stepExecutionDao;

    private ExecutionContextDao ecDao;

    private StepExecutionLogDao executionLogDao;

    /**
     * Provide default constructor with low visibility in case user wants to use
     * use aop:proxy-target-class="true" for AOP interceptor.
     */
    SimpleJobRepository() {
    }

    public SimpleJobRepository(JobInstanceDao jobInstanceDao, JobExecutionDao jobExecutionDao,
                               StepExecutionDao stepExecutionDao, ExecutionContextDao ecDao, StepExecutionLogDao logDao) {
        super();
        this.jobInstanceDao = jobInstanceDao;
        this.jobExecutionDao = jobExecutionDao;
        this.stepExecutionDao = stepExecutionDao;
        this.ecDao = ecDao;
        this.executionLogDao = logDao;
    }

    @Override
    public boolean isJobInstanceExists(String jobName, JobParameters jobParameters) {
        return jobInstanceDao.getJobInstance(jobName, jobParameters) != null;
    }

    @Override
    public JobExecution createJobExecution(String jobName, JobParameters jobParameters, int stepCount)
            throws JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {

        Assert.notNull(jobName, "Job name must not be null.");
        Assert.notNull(jobParameters, "JobParameters must not be null.");

		/*
         * Find all jobs matching the runtime information.
		 *
		 * If this method is transactional, and the isolation level is
		 * REPEATABLE_READ or better, another launcher trying to start the same
		 * job in another thread or process will block until this transaction
		 * has finished.
		 */

        JobInstance jobInstance = jobInstanceDao.getJobInstance(jobName, jobParameters);
        ExecutionContext executionContext;

        // existing job instance found
        if (jobInstance != null) {

            List<JobExecution> executions = jobExecutionDao.findJobExecutions(jobInstance);

            // check for running executions and find the last started
            for (JobExecution execution : executions) {
                if (execution.isRunning()) {
                    throw new JobExecutionAlreadyRunningException("A job execution for this job is already running: "
                            + jobInstance);
                }

                BatchStatus status = execution.getStatus();
                if (execution.getJobParameters().getParameters().size() > 0 && (status == BatchStatus.COMPLETED || status == BatchStatus.ABANDONED)) {
                    throw new JobInstanceAlreadyCompleteException(
                            "A job instance already exists and is complete for parameters=" + jobParameters
                                    + ".  If you want to run this job again, change the parameters.");
                }
            }
            executionContext = ecDao.getExecutionContext(jobExecutionDao.getLastJobExecution(jobInstance));
        } else {
            // no job found, create one
            jobInstance = jobInstanceDao.createJobInstance(jobName, jobParameters);
            executionContext = new ExecutionContext();
        }

        JobExecution jobExecution = new JobExecution(jobInstance, jobParameters, null, stepCount);
        jobExecution.setExecutionContext(executionContext);
        jobExecution.setLastUpdated(new Date(System.currentTimeMillis()));
        if (jobParameters.getString(JobExecution.JOB_OWNER_ID) != null) {
            jobExecution.setOwnerId(jobParameters.getString(JobExecution.JOB_OWNER_ID));
        }
        if (jobParameters.getString(JobExecution.JOB_OWNER_NAME) != null) {
            jobExecution.setOwnerName(jobParameters.getString(JobExecution.JOB_OWNER_NAME));
        }
        if (jobParameters.getString(JobExecution.OWNER_JOB_ID) != null) {
            jobExecution.setOwnerJobId(jobParameters.getString(JobExecution.OWNER_JOB_ID));
        }

        // Save the JobExecution so that it picks up an ID (useful for clients
        // monitoring asynchronous executions):
        jobExecutionDao.saveJobExecution(jobExecution);
        ecDao.saveExecutionContext(jobExecution);

        return jobExecution;

    }

    @Override
    public void update(JobExecution jobExecution) {

        Assert.notNull(jobExecution, "JobExecution cannot be null.");
        Assert.notNull(jobExecution.getJobId(), "JobExecution must have a Job ID set.");
        Assert.notNull(jobExecution.getId(), "JobExecution must be already saved (have an id assigned).");

        jobExecution.setLastUpdated(new Date(System.currentTimeMillis()));

        jobExecutionDao.synchronizeStatus(jobExecution);
        jobExecutionDao.updateJobExecution(jobExecution);
    }

    @Override
    public void add(StepExecution stepExecution) {
        validateStepExecution(stepExecution);

        stepExecution.setLastUpdated(new Date(System.currentTimeMillis()));
        stepExecutionDao.saveStepExecution(stepExecution);
        ecDao.saveExecutionContext(stepExecution);
    }

    @Override
    public void addAll(Collection<StepExecution> stepExecutions) {
        Assert.notNull(stepExecutions, "Attempt to save a null collection of step executions");
        for (StepExecution stepExecution : stepExecutions) {
            validateStepExecution(stepExecution);
            stepExecution.setLastUpdated(new Date(System.currentTimeMillis()));
        }
        stepExecutionDao.saveStepExecutions(stepExecutions);
        ecDao.saveExecutionContexts(stepExecutions);
    }

    @Override
    public void update(StepExecution stepExecution) {
        validateStepExecution(stepExecution);
        Assert.notNull(stepExecution.getId(), "StepExecution must already be saved (have an id assigned)");

        stepExecution.setLastUpdated(new Date(System.currentTimeMillis()));
        stepExecutionDao.updateStepExecution(stepExecution);
        checkForInterruption(stepExecution);
    }

    private void validateStepExecution(StepExecution stepExecution) {
        Assert.notNull(stepExecution, "StepExecution cannot be null.");
        Assert.notNull(stepExecution.getStepName(), "StepExecution's step name cannot be null.");
        Assert.notNull(stepExecution.getJobExecutionId(), "StepExecution must belong to persisted JobExecution");
    }

    @Override
    public void updateExecutionContext(StepExecution stepExecution) {
        validateStepExecution(stepExecution);
        Assert.notNull(stepExecution.getId(), "StepExecution must already be saved (have an id assigned)");
        ecDao.updateExecutionContext(stepExecution);
    }

    @Override
    public void updateExecutionContext(JobExecution jobExecution) {
        ecDao.updateExecutionContext(jobExecution);
    }

    @Override
    public StepExecution getLastStepExecution(JobInstance jobInstance, String stepName) {
        List<JobExecution> jobExecutions = jobExecutionDao.findJobExecutions(jobInstance);
        List<StepExecution> stepExecutions = new ArrayList<StepExecution>(jobExecutions.size());

        for (JobExecution jobExecution : jobExecutions) {
            stepExecutionDao.addStepExecutions(jobExecution);
            for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
                if (stepName.equals(stepExecution.getStepName())) {
                    stepExecutions.add(stepExecution);
                }
            }
        }

        StepExecution latest = null;
        for (StepExecution stepExecution : stepExecutions) {
            if (latest == null) {
                latest = stepExecution;
            }
            if (latest.getStartTime().getTime() < stepExecution.getStartTime().getTime()) {
                latest = stepExecution;
            }
            // Use step execution ID as the tie breaker if start time is identical
            if (latest.getStartTime().getTime() == stepExecution.getStartTime().getTime() &&
                    latest.getId() < stepExecution.getId()) {
                latest = stepExecution;
            }
        }

        if (latest != null) {
            ExecutionContext stepExecutionContext = ecDao.getExecutionContext(latest);
            latest.setExecutionContext(stepExecutionContext);
            ExecutionContext jobExecutionContext = ecDao.getExecutionContext(latest.getJobExecution());
            latest.getJobExecution().setExecutionContext(jobExecutionContext);
        }

        return latest;
    }

    @Override
    public int abandonLongTimeJobExecution(int interval) {
        return jobExecutionDao.abandonJobExecutedTimeLt(interval);
    }

    @Override
    public PageData<JobExecution> getJobExecution(int page, int size) {
        return jobExecutionDao.findJobExecutions(page, size);

    }

    @Override
    public PageData<JobExecution> getFailedJobExecution(int page, int size) {
        return jobExecutionDao.findFailedJobExecutions(page, size);

    }

    @Override
    public PageData<JobExecution> getUnExceptedJobExecution(int page, int size) {
        return jobExecutionDao.findUnExcepedJobExecutions(page, size);
    }

    @Override
    public PageData<JobExecution> getLongTimeJobExecution(int page, int size, int minutes) {
        return jobExecutionDao.findLongTimeJobExecutions(page, size, minutes);
    }

    @Override
    public List<StepExecution> getStepExecutions(JobExecution jobExecution) {
        return stepExecutionDao.getStepExecutions(jobExecution);
    }

    /**
     * @return number of executions of the step within given job instance
     */
    @Override
    public int getStepExecutionCount(JobInstance jobInstance, String stepName) {
        int count = 0;
        List<JobExecution> jobExecutions = jobExecutionDao.findJobExecutions(jobInstance);
        for (JobExecution jobExecution : jobExecutions) {
            stepExecutionDao.addStepExecutions(jobExecution);
            for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
                if (stepName.equals(stepExecution.getStepName())) {
                    count++;
                }
            }
        }
        return count;
    }

    /**
     * Check to determine whether or not the JobExecution that is the parent of
     * the provided StepExecution has been interrupted. If, after synchronizing
     * the status with the database, the status has been updated to STOPPING,
     * then the job has been interrupted.
     *
     * @param stepExecution
     */
    private void checkForInterruption(StepExecution stepExecution) {
        JobExecution jobExecution = stepExecution.getJobExecution();
        jobExecutionDao.synchronizeStatus(jobExecution);
        if (jobExecution.isStopping()) {
            logger.info("Parent JobExecution is stopped, so passing message on to StepExecution");
            stepExecution.setTerminateOnly();
        }
    }

    @Override
    public JobExecution getLastJobExecution(String jobName, JobParameters jobParameters) {
        JobInstance jobInstance = jobInstanceDao.getJobInstance(jobName, jobParameters);
        if (jobInstance == null) {
            return null;
        }
        JobExecution jobExecution = jobExecutionDao.getLastJobExecution(jobInstance);

        if (jobExecution != null) {
            jobExecution.setExecutionContext(ecDao.getExecutionContext(jobExecution));
            stepExecutionDao.addStepExecutions(jobExecution);
        }
        return jobExecution;
    }

    @Override
    public JobExecution getJobExecution(Long executionId) {
        return jobExecutionDao.getJobExecution(executionId);
    }

    @Override
    public JobInstance createJobInstance(String jobName, JobParameters jobParameters) {
        Assert.notNull(jobName, "A job name is required to create a JobInstance");
        Assert.notNull(jobParameters, "Job parameters are required to create a JobInstance");

        JobInstance jobInstance = jobInstanceDao.createJobInstance(jobName, jobParameters);

        return jobInstance;
    }

    @Override
    public JobExecution createJobExecution(JobInstance jobInstance,
                                           JobParameters jobParameters, String jobConfigurationLocation, int stepCount) {

        Assert.notNull(jobInstance, "A JobInstance is required to associate the JobExecution with");
        Assert.notNull(jobParameters, "A JobParameters object is required to create a JobExecution");

        JobExecution jobExecution = new JobExecution(jobInstance, jobParameters, jobConfigurationLocation, stepCount);
        ExecutionContext executionContext = new ExecutionContext();
        jobExecution.setExecutionContext(executionContext);
        jobExecution.setLastUpdated(new Date(System.currentTimeMillis()));

        // Save the JobExecution so that it picks up an ID (useful for clients
        // monitoring asynchronous executions):
        jobExecutionDao.saveJobExecution(jobExecution);
        ecDao.saveExecutionContext(jobExecution);

        return jobExecution;
    }

    @Override
    public Long getErrorLogCount(List<Long> executionId) {
        if (CollectionUtils.isEmpty(executionId)) {
            return 0l;
        }
        return executionLogDao.getErrorLogCount(executionId);
    }

    @Override
    public Long getWarningLogCount(List<Long> executionId) {
        if (CollectionUtils.isEmpty(executionId)) {
            return 0l;
        }
        return executionLogDao.getWarningLogCount(executionId);
    }

    @Override
    public PageData<StepExecutionLog> loadFirstPageLog(List<Long> executionId, int pageSize, long lastLogId) {
        PageData<StepExecutionLog> executionLogPageData = executionLogDao.loadLastestFirstPageLogs(executionId, pageSize, lastLogId);
        return getStepExecutionLogPageData(executionId, executionLogPageData);
    }

    @Override
    public PageData<StepExecutionLog> loadFirstPageLogNoTrace(List<Long> executionId, int pageSize, long lastLogId) {
        PageData<StepExecutionLog> executionLogPageData = executionLogDao.loadLastestFirstPageLogsNoTrace(executionId, pageSize, lastLogId);
        return getStepExecutionLogPageData(executionId, executionLogPageData);
    }

    @Override
    public PageData<StepExecutionLog> loadFirstPageLogOnlyError(List<Long> executionId, int pageSize, long lastLogId) {
        PageData<StepExecutionLog> executionLogPageData = executionLogDao.loadLastestFirstPageLogsOnlyError(executionId, pageSize, lastLogId);
        return getStepExecutionLogPageData(executionId, executionLogPageData);
    }

    private PageData<StepExecutionLog> getStepExecutionLogPageData(List<Long> executionId, PageData<StepExecutionLog> executionLogPageData) {
        if (!CollectionUtils.isEmpty(executionId)) {
            //判断是否是执行完的状态
            Long lastId = executionId.get(executionId.size() - 1);
            JobExecution jobExecution = jobExecutionDao.getJobExecution(lastId);
            if (jobExecution == null) {
                return new PageData<>(executionLogPageData.getTotalPages(), executionLogPageData.getTotalElements(), executionLogPageData.getContent(), true);
            }
            BatchStatus batchStatus = jobExecution.getStatus();
            if (BatchStatus.COMPLETED.equals(batchStatus)
                    || BatchStatus.ABANDONED.equals(batchStatus)
                    || BatchStatus.FAILED.equals(batchStatus)
                    || BatchStatus.STOPPED.equals(batchStatus)) {
                return new PageData<>(executionLogPageData.getTotalPages(), executionLogPageData.getTotalElements(), executionLogPageData.getContent(), true);
            }
        }
        return executionLogPageData;
    }

    @Override
    public PageData<StepExecutionLog> loadLogs(List<Long> executionId, int pageSize, long lastLogId) {
        return executionLogDao.loadLogs(executionId, pageSize, lastLogId);
    }

    @Override
    public PageData<StepExecutionLog> loadLogsNoTrace(List<Long> executionId, int pageSize, long lastLogId) {
        return executionLogDao.loadLogsNoTrace(executionId, pageSize, lastLogId);
    }

    @Override
    public PageData<StepExecutionLog> loadLogsOnlyError(List<Long> executionId, int pageSize, long lastLogId) {
        return executionLogDao.loadLogsOnlyError(executionId, pageSize, lastLogId);
    }

    @Override
    public void trace(SimpleStepContext simpleStepContext, String title, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.TRACE, simpleStepContext.getJobId(), simpleStepContext.getStepId(), title, content);
        executionLogDao.log(log);
    }

    @Override
    public void info(SimpleStepContext simpleStepContext, String title, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.INFO, simpleStepContext.getJobId(), simpleStepContext.getStepId(), title, content);
        executionLogDao.log(log);
    }

    @Override
    public void warning(SimpleStepContext simpleStepContext, String title, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.WARNING, simpleStepContext.getJobId(), simpleStepContext.getStepId(), title, content);
        executionLogDao.log(log);
    }

    @Override
    public void error(SimpleStepContext simpleStepContext, String title, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.ERROR, simpleStepContext.getJobId(), simpleStepContext.getStepId(), title, content);
        executionLogDao.log(log);
    }

    @Override
    public void trace(SimpleStepContext simpleStepContext, String titleKey, String title, String contentKey, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.TRACE, simpleStepContext.getJobId(), simpleStepContext.getStepId(), null, title, content, titleKey, contentKey);
        executionLogDao.log(log);
    }

    @Override
    public void info(SimpleStepContext simpleStepContext, String titleKey, String title, String contentKey, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.INFO, simpleStepContext.getJobId(), simpleStepContext.getStepId(), null, title, content, titleKey, contentKey);
        executionLogDao.log(log);
    }

    @Override
    public void warning(SimpleStepContext simpleStepContext, String titleKey, String title, String contentKey, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.WARNING, simpleStepContext.getJobId(), simpleStepContext.getStepId(), null, title, content, titleKey, contentKey);
        executionLogDao.log(log);
    }

    @Override
    public void error(SimpleStepContext simpleStepContext, String titleKey, String title, String contentKey, String content) {
        StepExecutionLog log = new StepExecutionLog(StepExecutionLog.LogLevel.ERROR, simpleStepContext.getJobId(), simpleStepContext.getStepId(), null, title, content, titleKey, contentKey);
        executionLogDao.log(log);
    }

    @Override
    public void logInBatch(SimpleStepContext simpleStepContext, List<CalcLog> logs) {
        List<StepExecutionLog> executionLogs = logs.stream().map(log ->
                new StepExecutionLog(log.getLogLevel(), simpleStepContext.getJobId(), simpleStepContext.getStepId(), log.getTitle(), log.getContent())).collect(Collectors.toList());
        executionLogDao.log(executionLogs);
    }
}